package online.longzipeng.mywebdemo;

import cn.hutool.core.collection.CollUtil;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import lombok.extern.slf4j.Slf4j;
import online.longzipeng.mywebdemo.entity.User;
import online.longzipeng.mywebdemo.mapper.UserMapper;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author lzp
 * @Date:2023/3/2
 * @description: 一些批处理操作思路， 可作为模板用于后续的批处理代码
 */
@Slf4j
@SpringBootTest(classes = MyWebDemoApplication.class,
		// 配置端口启动，否则获取失败
		webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
public class BatchDemo {

	@Autowired
	private UserMapper userMapper;

	/**
	 * 分页查询处理，减少单次批量处理的数据量级
	 */
	@Test
	public void test1() {
		// 预定义参数
		int page = 0;
		int pageSize = 5000;
		// 获取总数
		Integer total = userMapper.selectCount(null);
		// 计算页数
		int pages = total / pageSize;
		if (total % pageSize > 0) {
			pages++;
		}

		// 开始遍历处理数据
		for (; page < pages; page++) {
			List<User> users = userMapper.selectList(Wrappers.<User>lambdaQuery().last(String.format("LIMIT %s,%s", page * pageSize, pageSize)));

			users.forEach(user -> {
				/// 进行一些数据组装操作
			});
			/// 批量 修改/插入 操作


			User lastUser = users.get(users.size() - 1);
			log.info("最后一个要处理的用户的ID为：{}，名字：{}", lastUser.getId(), lastUser.getNickName());
		}

	}

	/**
	 * 未知总数的写法
	 */
	@Test
	public void test2() {
		// 预定义参数
		int page = 0;
		int pageSize = 500;
		// 开始遍历处理数据
		for (; ; ) {
			List<User> users = userMapper.selectList(Wrappers.<User>lambdaQuery().last(String.format("LIMIT %s,%s", (page++) * pageSize, pageSize)));
			users.forEach(user -> {
				/// 进行一些数据组装操作
			});
			/// 批量 修改/插入 操作

			if (CollUtil.isNotEmpty(users)) {
				User lastUser = users.get(users.size() - 1);
				log.info("最后一个要处理的用户的ID为：{}，名字：{}", lastUser.getId(), lastUser.getNickName());
			}

			if (users.size() < pageSize) {
				break;
			}
		}

	}

	/**
	 * 补充亿点点日志
	 */
	@Test
	public void test3() {
		// 预定义参数
		int page = 1;
		int pageSize = 500;
		// 获取总数
		Integer total = userMapper.selectCount(null);
		// 计算页数
		int pages = total / pageSize;
		if (total % pageSize > 0) {
			pages++;
		}

		// 总处理条数
		int count = 0;
		// 成功处理数
		int countOk = 0;
		// 处理失败记录
		List<Integer> wrongIds = new ArrayList<>();
		// 已过分钟数
		int countMinute = 1;

		long start = System.currentTimeMillis();
		// 开始遍历处理数据
		log.info("================== 开始批量处理数据 ==================");
		log.info("待处理条数：{}", total);
		log.info("总页数：{}", pages);
		log.info("每页条数：{}", pageSize);
		for (; page < pages; page++) {
			log.info("================== 当前进度{}/{} ==================", page, pages);
			List<User> users = userMapper.selectList(Wrappers.<User>lambdaQuery().last(String.format("LIMIT %s,%s", (page - 1) * pageSize, pageSize)));
			for (User user : users) {
				/// 进行一些数据组装操作
				if (user.getId() % 99 == 0) {
					wrongIds.add(user.getId());
				} else {
					countOk++;
				}
				count++;

				/// 模拟耗时操作
				try {
					Thread.sleep(10);
				} catch (InterruptedException e) {
					throw new RuntimeException(e);
				}

				// 每分钟输出一次日志，方便演示，这里是每10s输出一次
				if ((System.currentTimeMillis() - start) / 1000 / 10 > countMinute) {
					log.info("已耗时：{} s", (System.currentTimeMillis() - start) / 1000);
					log.info("当前总条数：{}", count);
					log.info("处理成功数：{}", countOk);
					log.info("处理失败数：{}", wrongIds.size());
					log.info("当前处理用户信息：{} : {}", user.getId(), user.getNickName());
					countMinute++;
				}
			}
			/// 批量 修改/插入 操作
			log.info("已耗时：{} s", (System.currentTimeMillis() - start) / 1000);
			log.info("当前总条数：{}", count);
			log.info("处理成功数：{}", countOk);
			log.info("处理失败数：{}", wrongIds.size());
			if (CollUtil.isNotEmpty(users)) {
				User user = users.get(users.size() - 1);
				log.info("{} : {}", user.getNickName(), user.getId());
			}
		}
		log.info("========================== 运行完毕 ==========================");
		log.info("总耗时：{} s", (System.currentTimeMillis() - start) / 1000);
		log.info("总处理条数：{}", count);
		log.info("处理成功数：{}", countOk);
		log.info("处理失败数：{}", wrongIds.size());
		log.info("处理失败数据id集合：{}", wrongIds);
	}

	/**
	 * 多线程优化查询，【切数据版 ，按线程数量切割数据，直接处理】
	 * + 需要程序进行大量计算
	 * + 数据库能承受较大并发
	 * + 多核CPU才能真正意义上的并行，不然就是宏观并行，微观串行 o(╥﹏╥)o
	 */
	@Test
	public void test4() {
		// 预定义参数
		int threadNum = 5;

		long start = System.currentTimeMillis();
		// 获取总数
		Integer total = userMapper.selectCount(null);
		// 创建线程池，这里为了简便操作直接用Executors创建，推荐自行集成配置线程池
		ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
		// 设置信号标，用于等待所有线程执行完
		CountDownLatch countDownLatch = new CountDownLatch(threadNum);

		// 计算线程需要处理的数据量的递增步长
		int threadTotalStep = total / threadNum;
		// 判断是否有余数，如果有多出的数据，补给最后一个线程
		int more = total % threadNum;

		// 开启 threadNum 个线程处理数据
		for (int i = 0; i < threadNum; i++) {
			int finalI = i;
			executorService.execute(() -> {
				int current = threadTotalStep * finalI;
				int threadTotal = threadTotalStep;
				/// 如果有余数，最后一次计算得补充余数
				if (more > 0 && finalI == threadNum - 1) {
					threadTotal += more;
				}
				List<User> users = userMapper.selectList(Wrappers.<User>lambdaQuery().last(String.format("LIMIT %s,%s", current, threadTotal)));

				users.forEach(user -> {
					/// 进行一些数据组装操作

					/// 进行一些耗时操作
					try {
						Thread.sleep(1);
					} catch (InterruptedException e) {
						throw new RuntimeException(e);
					}
				});
				/// 批量 修改/插入 操作

				User user = users.get(users.size() - 1);
				log.info("线程-{} 处理的最后一个数据的id为：{}", finalI + 1, user.getId());
				countDownLatch.countDown();
			});

		}

		try {
			countDownLatch.await();
			executorService.shutdown();
			log.info("总耗时：{} s", (System.currentTimeMillis() - start) / 1000);
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
	}

	/**
	 * 多线程优化查询，【分页版1 ，先按数量切数据，再在每个线程中分页处理数据】
	 * + 需要程序进行大量计算
	 * + 数据库能承受较大并发
	 * + 多核CPU才能真正意义上的并行，不然就是宏观并行，微观串行 o(╥﹏╥)o
	 */
	@Test
	public void test5() {
		// 预定义参数
		int threadNum = 5; // 线程数
		int pageSize = 500; // 每页处理条数

		long start = System.currentTimeMillis();
		// 获取总数
		Integer total = userMapper.selectCount(null);
		// 创建线程池，这里为了简便操作直接用Executors创建，推荐自行集成配置线程池
		ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
		// 设置信号标，用于等待所有线程执行完
		CountDownLatch countDownLatch = new CountDownLatch(threadNum);

		// 计算线程需要处理的数据量的递增步长
		int threadTotalStep = total / threadNum;
		// 判断是否有余数，如果有多出的数据，补给最后一个线程
		int more = total % threadNum;

		// 开启 threadNum 个线程处理数据
		for (int i = 0; i < threadNum; i++) {
			int finalI = i;
			executorService.execute(() -> {
				/// 数据总数就是 数据总数步长
				int threadTotal = threadTotalStep;
				// 获取上一个线程最终行数
				int oldThreadCurrent = threadTotalStep * finalI;
				/// 如果有余数，最后一次计算得补充余数
				if (more > 0 && finalI == threadNum - 1) {
					threadTotal += more;
				}
				log.info("线程-{} 要处理的数据总数为：{}", finalI + 1, threadTotal);
				// 计算页数
				int pages = threadTotal / pageSize;
				if (threadTotal % pageSize > 0) {
					pages++;
				}
				// 统计数量，当等于线程总总数时退出循环，避免重复计数
				int handleCount = 0;
				// 获取最后一个user
				User lastUser = new User();
				// 开始遍历处理数据
				for (int page = 0; page < pages; page++) {
					List<User> users = userMapper.selectList(
							Wrappers.<User>lambdaQuery()
									.last(String.format("LIMIT %s,%s", page * pageSize + oldThreadCurrent, pageSize)));

					for (User user : users) {
						handleCount++;
						if (handleCount == threadTotal) {
							break;
						}

						/// 模拟真正的逻辑处理，耗时操作
						try {
							Thread.sleep(1);
						} catch (InterruptedException e) {
							throw new RuntimeException(e);
						}
					}

					/// 批量 修改/插入 操作

					if (CollUtil.isNotEmpty(users)) {
						lastUser = users.get(users.size() - 1);
					}
				}
				log.info("线程-{} 处理的最后一个数据的id为：{}", finalI + 1, lastUser.getId());
				countDownLatch.countDown();
			});

		}

		try {
			countDownLatch.await();
			executorService.shutdown();
			log.info("总耗时：{} s", (System.currentTimeMillis() - start) / 1000);
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}

	}

	/**
	 * 多线程优化查询，【分页版2】 我的初版写法，写的有点烂【不建议用】，仅用于梳理思路
	 */
	@Test
	public void test6() {
		// 预定义参数
		int pageSize = 500;
		int threadNum = 5;
		// 获取总数
		Integer total = userMapper.selectCount(null);
		// 计算页数
		int pages = total / pageSize;
		if (total % pageSize > 0) {
			pages++;
		}
		long start = System.currentTimeMillis();

		// 创建线程池
		ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
		// 设置信号标
		CountDownLatch countDownLatch = new CountDownLatch(threadNum);

		log.info("总页数：{}", pages);
		// 获取每个线程的分页基数增长步长 【考虑到存在有余数的情况，直接默认+1】
		int pageStep = pages / threadNum + 1;

		// 记录末位值，验证测试结果是否正确
		String[] nameArr = new String[threadNum];

		for (int i = 0; i < threadNum; i++) {
			int finalI = i;
			// 如果已经大于最大页数，就没必要开线程跑了
			if (pageStep * finalI > pages) {
				countDownLatch.countDown();
				break;
			}
			executorService.execute(() -> {
				int page = pageStep * finalI;
				int handleCount = 0;
				// 开始遍历处理数据
				for (; page < pageStep * (finalI + 1); page++) {
					List<User> users = userMapper.selectList(Wrappers.<User>lambdaQuery().last(String.format("LIMIT %s,%s", page * pageSize, pageSize)));
					if (CollUtil.isEmpty(users)) {
						break;
					}
					for (User user : users) {
						handleCount++;
						/// 进行一些耗时操作
						try {
							Thread.sleep(1);
						} catch (InterruptedException e) {
							throw new RuntimeException(e);
						}
					}
					/// 批量 修改/插入 操作

					User user = users.get(users.size() - 1);
					// 记录最后一个用户的名称
					nameArr[finalI] = user.getNickName();
				}
				log.info("=================线程 {} 跑完了啊sir，总处理数：{} =================", finalI, handleCount);
				countDownLatch.countDown();
			});
		}

		try {
			countDownLatch.await();
			executorService.shutdown();
			log.info("总耗时：{} s", (System.currentTimeMillis() - start) / 1000);
			log.info(Arrays.toString(nameArr));
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
	}

}
